/**
 * Copyright 2010 Nicholas Alexander Cross
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.dataporter.io.csv;

import org.apache.log4j.Logger;
import org.csvreader.CsvReader;
import org.dataporter.data.DataObject;
import org.dataporter.io.DataModel;
import org.dataporter.io.DataReader;
import org.dataporter.io.EntitySource;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class CSVDataReader<Source extends EntitySource<? extends InputStream>> implements DataReader<Source> {

    private static final Logger logger = Logger.getLogger( CSVDataReader.class );

    private final Map<String, CsvReader> readers = new HashMap<String, CsvReader>();
    private final DataModel model;

    private EntitySource<? extends InputStream> source;
    private CsvReader currentReader;
    private String currentEntityName;
    private Iterator<String> entityIterator;

    public CSVDataReader( DataModel model ) {
        this.model = model;
    }

    public void init( Source source ) throws Exception {
        this.source = source;
        this.entityIterator = source.entityNames().iterator();
    }

    private CsvReader readerForEntity(String entityName) {
        CsvReader reader = readers.get( entityName );

        if ( reader == null )
        {
            reader = new CsvReader( source.sourceForEntity(entityName), Charset.forName( "UTF-8" ) );
            readers.put( entityName, reader );
        }

        return reader;
    }

    public DataObject read() {
        try
        {
            DataObject dataObject = null;
            CsvReader reader = currentReader();

            if ( reader != null )
            {
                if ( reader.readRecord() )
                {
                    dataObject = new DataObject(
                            currentEntityName,
                            model.getEntity( currentEntityName ).getPrimaryKeyNames()
                    );

                    String[] headers = reader.getHeaders();
                    for ( String header : headers )
                    {
                        //reader next data object
                        dataObject.put( header, reader.get( header ) );
                    }
                }
                else if (entityIterator.hasNext()) {
                    //read from the next source
                    currentReader = null;
                    currentEntityName = null;
                    return read();
                }
            }

            return dataObject;
        }
        catch ( IOException e )
        {
            logger.error( "Failed to read from csv file", e );
            return null;
        }
    }

    private CsvReader currentReader() throws IOException {
        if (currentReader == null) {
            if (entityIterator.hasNext()) {
                currentEntityName = entityIterator.next();
                currentReader = readerForEntity( currentEntityName );
                currentReader.readHeaders();
            }
        }

        return currentReader;
    }

    public void close() {
        Collection<CsvReader> allReaders = readers.values();

        for (CsvReader reader : allReaders) {
            try
            {
                reader.close();
            }
            catch ( Exception e )
            {
                logger.error( "Failed to close reader", e );
            }
        }

        readers.clear();
    }

}
